# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: helpers\issue_tracker\clients\github_client.py
import json, logging
from . import util
from ..exceptions import ConfigError
from .base_client import BaseClient
from requests.auth import HTTPBasicAuth

logger = logging.getLogger(__name__)


class GithubClient(BaseClient):
    """
    GithubClient handles all communication between Acunetix and Github API.
    
    The current client implementation works on version 3 of Github's API and
    supports the following functionality:
        - Authentication
            (a). Basic Authentication
            .. note: Github API supports two forms of OAuth authentication,
                     neither of which are currently supported
    
        - Get Projects(*)
        - Get Issue Status(*)
        - Get Issue Types
            + For all projects
            + For specified project
        - Create Issue
            .. note: Creating bulk issues currently not supported, may be implemented
    
        * In the context of the authenticated user
    
    Instantiation of the Github client should follow the following structure:
    
    >>> import helpers.issue_tracker.clients.client as clients
    >>>
    >>> target_config = {
    >>>     "bug_tracker": "github",
    >>>         "url": "https://api.github.com/",
    >>>         "auth": {
    >>>             "kind": "http_basic",
    >>>             "user": "root",
    >>>             "password": "toor"
    >>>         },
    >>>         "project": 'acx/acx',
    >>>         "issuetype": "bug"  # Optional
    >>>     }
    >>>
    >>> client = clients.get_client(target_config)
    
    See the :func:`establish_connection` as to how establishing a connection works
    See the :func:`get_client` in :file:`client.py` as to how the factory function works
    """

    def __init__(self, target_config, proxies_cb=None, ssl_ca_bundle_path=True):
        super().__init__(target_config, proxies_cb, ssl_ca_bundle_path)
        self.session.headers.update({"Content-Type": "application/vnd.github.v3+json"})

    def _format_vulnerability_description(self, description):
        """Format Acunetix vulnerability description to Github markdown
        
        Currently Github's Markdown does not (by default) support colored text,
        thus all vulnerability data will be pushed in black and white.
        
        .. note::
        Currently works by using string swaps rather than regex. Tested this
        using regex however was visibly slower when working with large
        descriptions and the code itself was a big eyesore.
        
        :param description: Acunetix Vulnerability description in HTML format
        :return: Acunetix Vulnerability description in Github Markdown format
        """
        tags = [
            ("<i>", "*"),
            ("</i>", "*"),
            ("<code><pre>", "```\n"),
            ("</pre></code>", "\n```"),
            ("<pre><code>", "```\n"),
            ("</code></pre>", "\n```"),
            ("<br/>", "\n"),
            ("<hr>", "\n----\n"),
            ("<strong>", "**"),
            ("</strong>", "**"),
            ("<p>", "\n"),
            ("</p>", ""),
            ("<h1>", "\n# "),
            ("</h1>", "\n"),
            ("<h2>", "\n## "),
            ("</h2>", "\n"),
            ("<h3>", "\n### "),
            ("</h3>", "\n"),
            ("<h4>", "\n#### "),
            ("</h4>", "\n"),
            ("<h5>", "\n##### "),
            ("</h5>", "\n"),
            ("<h6>", "\n###### "),
            ("</h6>", "\n"),
            ("<code>", "```\n"),
            ("</code>", "\n```"),
            ('<span class="bb-dark">', ""),
            ("</span>", ""),
            ('<div class="bb-coolbox">', "> "),
            ("</div>", ""),
            ("<ol>", "\n"),
            ("</ol>", "\n"),
            ("<ul>", "\n"),
            ("</ul>", "\n"),
            ("<li>", " + "),
            ("</li>", ""),
            ("<pre>", ""),
            ("</pre>", ""),
        ]
        for tag in tags:
            description = description.replace(tag[0], tag[1])

        return description

    def _format_vulnerability_references(self, references):
        """Format vulnerability references to Github links
        
        .. note:
        
        Vulnerability references are either paired in two, reference title
         and reference URL. Otherwise they're mapped in singular, URL only.
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>>     }
        >>>
        >>> references = [['Acunetix SQL Injection Attack', 'http://www.acunetix.com/...'],
        >>>               ['VIDEO: SQL Injection tutorial', 'http://www.acunetix.com/blog/...']]
        >>>
        >>> github_client = clients.get_client(target_config)
        >>> github_client._format_vulnerability_references(references)
        >>> ['[Acunetix SQL Injection Attack(http://www.acunetix.com/...)]',
        >>>  '[VIDEO: SQL Injection tutorial(http://www.acunetix.com/blog/...)]']
        
        :param references: Vulnerability reference in a 2-dimensional list
        :return: New 1-dimensional list containing formatted references
        """
        html_references = []
        try:
            for pair in references:
                if not isinstance(pair, (list, tuple)):
                    raise AssertionError("Invalid type, must be of type list or tuple")
                if len(pair) == 1:
                    html_references.append("[" + pair[0] + "]")
                elif len(pair) == 2:
                    html_references.append("[" + pair[0] + "](" + pair[1] + ")")
                else:
                    raise ValueError(
                        "Reference pair has too many or too few"
                        + "items to unpack, must be in pairs of 1 or 2"
                    )

        except (AttributeError, TypeError):
            raise AssertionError("References must be an iterable object")

        return html_references

    def _establish_session(self):
        """Establish a new user session with Github
        
        .. note:: Currently only supports HTTP Basic Authentication
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>>     }
        >>> github_client = clients.get_client(target_config)
        
        :return: True if authentication successful, false if otherwise.
        """
        if self.target_config["auth"]["kind"] == "http_basic":
            request = super()._request(
                endpoint="/", auth=HTTPBasicAuth(self.username, self.password)
            )
        else:
            raise ConfigError(
                "Unknown Authentication method parsed - Must be 'http_basic'."
            )
        if request.status_code != 200:
            raise ConfigError("Invalid username and password or account locked out.")
        try:
            if not request.headers.get("X-GitHub-Request-Id"):
                if "application/json" not in request.headers["Content-Type"]:
                    if "current_user_url" not in request.text:
                        raise KeyError
        except KeyError:
            raise ConfigError(
                "Unexpected response returned. Normally caused by an invalid URL."
            )

    def add_vulnerability(self, vuln_data):
        """POST new issue into Github repository with specified data.
        
        Vulnerability description between HTML to Markdown format is
        handled automatically. Do NOT modify the Acunetix vulnerability
        descriptions.
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> vuln_data = {
        >>>    'vt_name': 'foo',
        >>>    'long_description': '',
        >>>    'description': 'Test Description',
        >>>    'details': 'Attack Details',
        >>>    'vt_tags': ['sql_injection', 'CWE-89']
        >>> }
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>>
        >>> github_client = clients.get_client(target_config)
        >>>
        >>> github_client.add_vulnerability(vuln_data)
        
        :param vuln_data: Issue data POSTed to Github. Format:
                            >>> {
                            >>>     'title': 'Acunetix',
                            >>>     'body': 'Vulnerability Description',
                            >>>     'labels': ['CWE-00', 'sql_injection']
                            >>> }
        :return: New issue ID if successfully created, otherwise None
        """
        repo = self.target_config["project"]["project_name"]
        vuln_description = vuln_data["description"]
        attack_details = vuln_data["details"]
        request_data = vuln_data.get("request")
        if request_data:
            request_details = "\t" + vuln_data["request"]
            request_details = request_details.replace("\n", "\n\t")
        else:
            request_details = ""
        target_information = (
            "| **Target URL** | "
            + vuln_data.get("target_address")
            + "|\n"
            + "------------ | -------------\n"
            + "| **Severity**   | "
            + util.evaluate_severity(vuln_data["severity"])
            + "|\n"
        )
        recommendation = (
            vuln_data["recommendation"] if vuln_data["recommendation"] else ""
        )
        impact = vuln_data["impact"] if vuln_data["impact"] else ""
        description = self._format_vulnerability_description(
            target_information
            + "<h5>Vulnerability Description</h5>"
            + vuln_description
            + "\n<h5>Attack Details</h5>"
            + attack_details
            + "\n<h5>HTTP Request</h5>"
            + request_details
            + "\n<h5>Impact</h5>"
            + impact
            + "\n<h5>Remediation</h5>"
            + recommendation
        )
        reference_tags = self._format_vulnerability_references(vuln_data["refs"])
        if len(reference_tags) > 0:
            description += "\n<hr><h5>References:</h5>"
            for reference_tag in reference_tags:
                description += reference_tag + "<br/>"

            description += "<br/>"
        labels = super()._format_vulnerability_tags(vuln_data["vt_tags"])
        if "issue_type" in self.target_config.keys():
            labels.append(self.target_config["issue_type"]["issue_id"])
        post_data = json.dumps(
            {
                "title": "Acunetix - " + vuln_data["vt_name"],
                "body": self._format_vulnerability_description(description),
                "labels": labels,
            }
        ).encode("utf-8")
        request = super()._request(
            endpoint="/repos/" + repo + "/issues",
            auth=HTTPBasicAuth(self.username, self.password),
            data=post_data,
        )
        try:
            if request.status_code == 201:
                return request.json()["number"]
            raise Exception(
                "Invalid status code returned. Unable to add issue successfully."
            )
        except KeyError:
            raise Exception(
                "Unexpected response body returned, unable to parse issue number."
            )

    def get_projects(self):
        """Pull list of projects relative to authenticated user
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>> github_client = clients.get_client(target_config)
        >>>
        >>> github_client.get_projects()
        >>> ['test']
        
        :return: List of available projects (project's full name only)
        """
        request = super()._request(
            endpoint="/user/repos", auth=HTTPBasicAuth(self.username, self.password)
        )
        if request.status_code == 200:
            try:
                projects = request.json()
                return [
                    {
                        "project_id": str(project["id"]),
                        "project_name": project["full_name"],
                    }
                    for project in projects
                ]
            except KeyError:
                raise Exception(
                    "Received unexpected response body, unable to parse project name and ID."
                )

            raise Exception(
                "Request to Github unsuccessful. Unable to retrieve list of projects."
            )

    def get_issue_types(self):
        """
        Extract issue-types for every available project or all the issue
         types of a specific project.
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>>     }
        >>> github_client = clients.get_client(target_config)
        >>>
        >>> github_client.get_issue_types()
        >>> {"projects": ["test": {"issuetypes": ["bug", "task"]}]}
        >>>
        >>> github_client.get_issue_types(repository='test')
        >>> {"test": {"issuetypes": ["bug", "task"]}}
        
        :return: List of issues types per project for authenticated user.
        """
        repository = self.target_config["project"]["project_name"]
        request = super()._request(
            endpoint="/repos/" + repository + "/labels",
            auth=HTTPBasicAuth(self.username, self.password),
        )
        if request.status_code == 200:
            issue_types = [
                {"issue_id": label["name"], "issue_name": label["name"]}
                for label in request.json()
            ]
        else:
            issue_types = None
        return issue_types

    def get_issue_status(self, issue_id):
        """Get the status of a specific issue (e.g. done, closed)
        
        >>> import helpers.issue_tracker.clients.client as clients
        >>>
        >>> vuln_data = {
        >>>    'vt_name': 'foo',
        >>>    'long_description': '',
        >>>    'description': 'Test Description',
        >>>    'details': 'Attack Details',
        >>>    'vt_tags': ['sql_injection', 'CWE-89']
        >>> }
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "github",
        >>>         "url": "https://api.github.com/",
        >>>         "auth": {
        >>>             "kind": "http_basic",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>>
        >>> github_client = clients.get_client(target_config)
        >>>
        >>> issue_id = github_client.add_vulnerability(vuln_data)
        >>> github_client.get_issue_status(issue_id)
        >>> 'done'
        
        :param issue_id: ID of issue to extract status from
        :return: Status of the specified issue
        """
        repo = self.target_config["project"]["project_name"]
        request = super()._request(
            endpoint="/repos/" + repo + "/issues/" + str(issue_id),
            auth=HTTPBasicAuth(self.username, self.password),
        )
        try:
            if request.status_code == 200:
                return request.json()["state"]
        except KeyError:
            raise Exception("Unexpected response body returned")

        raise Exception("Unable to retrieve issue status for issue: " + str(issue_id))
